Still need to add tests.

LGTM but the CI is failing, and this reminds me that it would be good to add tests at least for the receivers that actually do something when close() is called.

Yes, I'm already working on tests in this same branch, just made an early PR because the code is ready for review.

Maybe to make it play better with Service we can keep close() as you implemented it, as a "graceful" shutdown, and when we introduce Service we can use cancel() to drop the current buffers. Then maybe stop() can take an option to decide if close() or cancel() is called (stop is usually implemented as self.cancel(); await self).

I think stop() doesn't make sense for receivers, because it is not a task. Idiomatically, we should have stop() only for things that look like there must be an activity there. So it is great for Actor, etc. But for a stream, I like close() better.

I think stop() doesn't make sense for receivers, because it is not a task. Idiomatically, we should have stop() only for things that look like there must be an activity there. So it is great for Actor, etc. But for a stream, I like close() better.

Mmm, receiver don't need to have task, but many do (like Merger, Timer and FileWatcher). If we don't add a task-like interface, it means we can't stop these receivers using the Receiver interface, which is the only way to access them if we apply a filter or map to them for example.

But maybe there are other solutions to this. We can discuss this when the time comes.

Please let me know when this is ready for a final review :)

Mmm, receiver don't need to have task, but many do (like Merger, Timer and FileWatcher). If we don't add a task-like interface, it means we can't stop these receivers using the Receiver interface, which is the only way to access them if we apply a filter or map to them for example.

That doesn't sound like an issue, because people still call close() on the receiver, and if the underlying thing is a FileWatcher for example, the close() would just call the stop(), which is what this PR is doing already.

And calling close() on Map and Filter would call close() of the underlying receiver. This PR has that too.

That doesn't sound like an issue, because people still call close() on the receiver, and if the underlying thing is a FileWatcher for example, the close() would just call the stop(), which is what this PR is doing already.

But stop() is async, and it is important to be able to wait until the thing actually stopped to avoid those annoying warnings in test we were talking about 😆

If we make close() async, then fine, but then we are ACKing that receivers can have background running tasks, and then why shouldn't they follow the Service interface?

This is ready, the only big change is that Receiver.close() is now async.

Last push to have a (Background)Service-compatible interface.

If we make close() async, then fine, but then we are ACKing that receivers can have background running tasks, and then why shouldn't they follow the Service interface?

BTW, the method can't be abstract, otherwise this is a breaking change. We need to provide a default empty implementation I guess. We can create an issue to make the method abstract for 2.0.

I've made the close() method not abstract and not async. With these,

• it is no longer a breaking change
• it becomes a signal to close(), in keeping with the idioms of the asyncio library, which was the behaviour already even though it was async, and not a "ensure it is closed before returning".

With this the two above concerns should also go away.

Rebased on latest

@llucax could you take a look at this again?